package com.tony.reactiveStream;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

/**
 * @version V1.0  </br>
 * @ClassName MySubscriber </br>
 * @Description TODO  </br>
 * @Author tonywang</br>
 * @DATE 3/14/2019 4:21 PM</br>
 */
public class MySubscriber implements Subscriber<Employee> {
    private Subscription subscription;

    private int counter = 0;

    @Override
    public void onSubscribe(Subscription s) {
        System.out.println("Subscribed");
        this.subscription = s;
        this.subscription.request(1);//请求消息
        System.out.println("onSubscribe requested 1 item");
    }

    @Override
    public void onNext(Employee pEmployee) {
        System.out.println("Processing Employee "+pEmployee);
        counter++;
        this.subscription.request(1);

    }

    @Override
    public void onError(Throwable t) {
        System.out.println("Some error happened");
        t.printStackTrace();
    }

    @Override
    public void onComplete() {
        System.out.println("All Processing Done");
    }

    public int getCounter() {
        return counter;
    }
}
